package demo.db.hbase;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import demo.db.hbase.utils.HbaseUtils;

public class HbaseConfigDemo {

    private static final Logger logger = LoggerFactory.getLogger(HbaseConfigDemo.class);

    public static TableName tableName = TableName.valueOf("t1");
    public static byte[] family = Bytes.toBytes("cf1");

    private static String quorum = "192.168.60.12:2181,192.168.60.25:2181,192.168.60.26:2181,192.168.60.37:2181,192.168.60.38:2181,192.168.60.39:2181,192.168.60.40:2181";
    private static Configuration configuration;

//    public static void main(String[] args) {
//        try {
//            demoTableName();
//        } catch (Exception e) {
//            logger.error("", e);
//        }
//    }

    static {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", quorum);
    }

    @Test
    public void demoTableName() {
        TableName tableName = TableName.valueOf("sn:tableName");
        logger.info(tableName.getNameAsString());
        logger.info(tableName.getNamespaceAsString());
        logger.info(tableName.getNameWithNamespaceInclAsString());
        logger.info(tableName.getQualifierAsString());
    }

    /** 删除表 */
    static void demoDropTable() throws IOException {
        // 表名
        String tableName = "test-hbase";
        try (Connection connection = ConnectionFactory.createConnection(configuration);
                // 表管理类
                Admin admin = connection.getAdmin();) {
            // 首先禁用表
            admin.disableTable(TableName.valueOf(tableName));
            // 最后删除表
            admin.deleteTable(TableName.valueOf(tableName));
        }
        logger.info("执行成功");
    }

    /** 创建表 */
    static void demoCreateTable() throws IOException {
        // 表名
        String tableName = "test-hbase";
        // 列族名
        String columnName = "info";
        try (Connection connection = ConnectionFactory.createConnection(configuration);
                // 表管理类
                Admin admin = connection.getAdmin();) {
            // 定义表名
            HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
            tableDescriptor.addFamily(new HColumnDescriptor(columnName));
            admin.createTable(tableDescriptor);
            // 定义表结构
            // HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnName);
            // admin.addColumn(TableName.valueOf(tableName), columnDescriptor);
        }

    }

    /** 扫描 */
    static void demoScan() throws IOException {
        // 开始行键
        String startRow = "0";
        // 结束行键
        String endRow = "9";

        try (Connection connection = ConnectionFactory.createConnection(configuration);
                Table table = connection.getTable(tableName);) {
            // 初始化Scan实例，并制定起始行
            Scan scan = new Scan().withStartRow(Bytes.toBytes(startRow)).withStopRow(Bytes.toBytes(endRow));
            // 增加过滤条件
            scan.addFamily(family);
            // 返回结果
            ResultScanner rs = table.getScanner(scan);
            // 迭代并取出结果
            for (Result result : rs) {
                HbaseUtils.printResult(result);
            }
        }
    }

    /** 删除单元格 */
    static void demoDelete() throws IOException {
        // 行键
        String rowkey = "1";
        // 列名
        String qulifer = "name";
        try (
                // 建立连接
                Connection connection = ConnectionFactory.createConnection(configuration);
                // 建立表连接
                Table table = connection.getTable(tableName);) {
            // 用行键来实例化Delete实例
            Delete delete = new Delete(Bytes.toBytes(rowkey));
            // 增加删除条件
            delete.addColumn(family, Bytes.toBytes(qulifer));
            // 执行删除
            table.delete(delete);
            logger.info("删除成功");
        }

    }

    /**
     * 创建连接耗时很长，平均18ms 。
     * <p>
     * Connection creation is a heavy-weight operation. Connection implementations
     * are thread-safe, so that the client can create a connection once, and share
     * it with different threads. Table and Admin instances, on the other hand, are
     * light-weight and are not thread-safe. Typically, a single connection per
     * client application is instantiated and every thread will obtain its own Table
     * instance. Caching or pooling of Table and Admin is not recommended.
     */
    @Test
    public void createConnectionCost() throws IOException {
        long start = System.currentTimeMillis();
        int i = 0;
        int n = 1_000;
        while (i < n) {
            try (Connection connection = ConnectionFactory.createConnection(configuration);) {
                i++;
            }
        }
        
        // 创建连接 1000 次，耗时：18497 ms.
        logger.info("创建连接 {} 次，耗时：{} ms.", n, System.currentTimeMillis() - start);
    }

    /**
     * connection.getTable 耗时时间很短，不到1毫秒。 The returned Table is not thread safe, a
     * new instance should be created for each using thread.
     */
    static void getTableCost() throws IOException {
        long start = System.currentTimeMillis();
        int i = 0;
        int n = 1_000;
        Connection connection = ConnectionFactory.createConnection(configuration);
        while (i < n) {
            try (Table table = connection.getTable(tableName);) {
                i++;
            }
        }
        connection.close();
        // 创建连接 1000 次，耗时：765 ms.
        logger.info("创建连接 {} 次，耗时：{} ms.", n, System.currentTimeMillis() - start);
    }

    /** 一次连接，先写，后读 */
    static void demoPutGetOneConnection() throws IOException {
        String rowKey = "1";
        Map<String, String> column = new HashMap<>();
        column.put("Name", "Stan");
        column.put("Age", "31");
        column.put("sex", "man");
        try (Connection connection = ConnectionFactory.createConnection(configuration);
                Table table = connection.getTable(tableName);) {
            Put put = new Put(Bytes.toBytes(rowKey));
            column.entrySet().forEach(e -> {
                put.addColumn(family, Bytes.toBytes(e.getKey()), Bytes.toBytes(e.getValue()));
            });
            table.put(put);
            Get get = new Get(Bytes.toBytes(rowKey));
            if (table.exists(get)) {
                Result result = table.get(get);
                logger.info("Result:{}", result);
                HbaseUtils.printResult(result);
            } else {
                logger.error("no data");
            }
        }
    }

    /** 插入 */
    static void demoPut() throws IOException {
        String rowKey = "1";
        Map<String, String> column = new HashMap<>();
        column.put("Name", "Stan");
        column.put("Age", "31");
        try (Connection connection = ConnectionFactory.createConnection(configuration);
                Table table = connection.getTable(tableName);) {
            Put put = new Put(Bytes.toBytes(rowKey));
            column.entrySet().forEach(e -> {
                put.addColumn(family, Bytes.toBytes(e.getKey()), Bytes.toBytes(e.getValue()));
            });
            table.put(put);
        }
    }

    /** 查询 */
    static void demoGet() throws IOException {
        String rowKey = "1";
        try (Connection connection = ConnectionFactory.createConnection(configuration);
                Table table = connection.getTable(tableName);) {
            Get get = new Get(Bytes.toBytes(rowKey));
            if (table.exists(get)) {
                Result result = table.get(get);
                logger.info("Result:{}", result);
                HbaseUtils.printResult(result);
            } else {
                logger.error("no data");
            }
        }
    }

}
